package SparkStreaming;

import com.alibaba.fastjson.JSONObject;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.sql.types.StructType;

/**
 * <pre>
 * Created with IntelliJ IDEA.
 * User: zhengzhi
 * Date: 2017/5/9
 * To change this template use File | Settings | File Templates.
 * </pre>
 */
public class TrackerDataRead {

    public static void main(String[] args) {
        SparkConf sparkConf = new SparkConf().setAppName("Test").setMaster("local[2]");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);
        SQLContext sqlContext = new SQLContext(sc);

//        String sql = "SELECT campaignUrn,count(*) as pv FROM trackerTestData group by campaignUrn";
//        JavaRDD<String> sourceData = sc.textFile("D:\\testData\\trackerTestData.txt");

        String sql1 = "SELECT os,COUNT(cookieId) as amount from trackerTestData where device='MOBILE' group by os";
        JavaRDD<String> sourceData1 = sc.textFile("D:\\testData\\phoneAmountTestData.txt");


//        sourceData.foreach(new VoidFunction<String>() {
//            @Override
//            public void call(String s) throws Exception {
//                System.out.println("@@@"+s);
//            }
//        });


        DataFrame dataFrame = sqlContext.jsonRDD(sourceData1);

//        System.out.println(dataFrame.toString());

        dataFrame.registerTempTable("trackerTestData");

        DataFrame allSiteCount = dataFrame.sqlContext().sql(sql1);

        JavaRDD<String> map = allSiteCount.toJavaRDD().map(new Function<Row, String>() {
            @Override
            public String call(Row row) throws Exception {
                StructType schema = row.schema();
                String[] fieldNames = schema.fieldNames();

                int length = row.length();
                JSONObject jsonObject = new JSONObject();

                for (int i = 0; i < length; i++) {
                    jsonObject.put(fieldNames[i], row.get(i));
                }
                return jsonObject.toJSONString();
            }
        });

        map.foreach(new VoidFunction<String>() {
                        @Override
                        public void call(String s) throws Exception {
                            System.out.println("@@@@@@@@@@@"+s);
                        }
                    }
        );


//        foreach(new VoidFunction<String>() {
//            @Override
//            public void call(String s) throws Exception {
//                System.out.println(s);
//            }
//        });
    }
}
